package com.atguigu.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @Auther yushupeng
 * @Data 2022/12/18
 */
// 数据流：web/app -> Nginx -> 日志服务器(.log) -> Flume -> Kafka(ODS) -> FlinkApp -> Kafka(DWD) -> FlinkApp -> Kafka(DWD)
// 程序:   Mock(lg.sh) -> Flume(f1) -> Kafka(ZK) -> BaseLogApp -> Kafka(ZK) -> DwdTrafficUserJumpDetail -> Kafka(ZK)
public class DwdTrafficUserJumpDetail {
    public static void main(String[] args) throws Exception {

        // TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.1 设置checkpoint
//        env.enableCheckpointing(5 * 6000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(10 * 6000L);
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000L));
//
//        //1.2 设置状态后端
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://192.168.60.202:8020/ck");
//        System.setProperty("HADOOP_USER", "atguigu");

        // TODO 2.读取Kafka 页面日志主题数据并创建流
        String topic = "dwd_traffic_page_log";
        String group = "dwd_traffic_user_jump_detail";
        DataStreamSource<String> kafkaDS = env.addSource(KafkaUtil.getFlinkKafkaConsumer(topic, group));

        //测试数据
        /*
        {"common":{"ar":"370000","ba":"Xiaomi","ch":"web","is_new":"0","md":"Xiaomi 10 Pro ","mid":"mid_2190279","os":"Android 11.0","uid":"688","vc":"v2.1.134"},"page":{"during_time":11863,"item":"34,27,14","item_type":"sku_ids","last_page_id":"cart","page_id":"trade"},"ts":1651303991000}

        {"common":{"ar":"370000","ba":"Xiaomi","ch":"web","is_new":"0","md":"Xiaomi 10 Pro ","mid":"mid_2190279","os":"Android 11.0","uid":"688","vc":"v2.1.134"},"page":{"during_time":11863,"item":"34,27,14","item_type":"sku_ids","page_id":"trade"},"ts":1651303991000}

        {"common":{"ar":"370000","ba":"Xiaomi","ch":"web","is_new":"0","md":"Xiaomi 10 Pro ","mid":"mid_2190279","os":"Android 11.0","uid":"688","vc":"v2.1.134"},"page":{"during_time":11863,"item":"34,27,14","item_type":"sku_ids","page_id":"trade"},"ts":1651304100000}

         */

        // TODO 3. 将每行数据转换为JSON对象
        SingleOutputStreamOperator<JSONObject> JSONObjectDS = kafkaDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String s) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(s);
                    return jsonObject;
                } catch (Exception e) {
                    System.out.println("脏数据" + s);
                    return null;
                }
            }
        });

        // TODO 4.提取事件时间&&按照Mid分组
        KeyedStream<JSONObject, String> keyedStream =
                JSONObjectDS.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject jsonObject, long l) {
                return jsonObject.getLong("ts");
            }
        }))
                .keyBy(JSON -> JSON.getJSONObject(
                        "common").getString("mid"));

        // TODO 5.定义CEP的模式序列
        Pattern<JSONObject, JSONObject> pattern =
                Pattern.<JSONObject>begin("first").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {
                String last_page_id = jsonObject.getJSONObject("page").getString("last_page_id");
                return last_page_id == null;
            }
        })
                .next("second").where(new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject) throws Exception {
                        String last_page_id = jsonObject.getJSONObject("page").getString("last_page_id");
                        return last_page_id == null;
                    }
                })
                .within(Time.seconds(10));


        // TODO 6.将模式序列作用到流上
        PatternStream<JSONObject> patternStream = CEP.pattern(keyedStream, pattern);

        // TODO 7.提取事件（匹配上的事件以及超时事件）
        // 超时事件存到侧输出流中
        OutputTag<String> timeoutTag = new OutputTag<String>("timeoutTag"){};

        SingleOutputStreamOperator<String> flatSelect = patternStream.flatSelect(
                timeoutTag,
                new PatternFlatTimeoutFunction<JSONObject, String>() {
                    @Override
                    public void timeout(Map<String, List<JSONObject>> map, long l, Collector<String> collector) throws Exception {
                        JSONObject element = map.get("first").get(0);
                        collector.collect(element.toJSONString());
                    }
                },
                new PatternFlatSelectFunction<JSONObject, String>() {
                    @Override
                    public void flatSelect(Map<String, List<JSONObject>> map, Collector<String> collector) throws Exception {
                        JSONObject element = map.get("first").get(0);
                        collector.collect(element.toJSONString());
                    }
                }
        );

        DataStream<String> sideOutput = flatSelect.getSideOutput(timeoutTag);
        // TODO 8.合并两个事件
        DataStream<String> unionStream = flatSelect.union(sideOutput);

        // TODO 9.将数据写到Kafka
        flatSelect.print("flatSelect>>>>>>>>>>");
        sideOutput.print("timeout>>>>>>>>>>>>>");
        String targetTopic = "dwd_traffic_user_jump_detail";
        unionStream.addSink(KafkaUtil.getFlinkKafkaProducer(targetTopic));

        // TODO 10.启动任务

        env.execute("DwdTrafficUserJumpDetail");
        
    }
}